package com.zyh.flink.day06.function;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

//通过计算用户24小时内，平均消费金额为例演示aggregateFunction
public class AggregateFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStreamSource = environment.socketTextStream("hadoop10", 9999);

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] ss = value.split("\\s+");
                out.collect(Tuple2.of(ss[0], Integer.valueOf(ss[1])));
            }
        }).keyBy(t -> t.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<String> result = windowedStream.aggregate(new MyAggregateFunction());

        result.print();

        environment.execute("aggregateFunction");
    }
}

/*
  AggregateFunction<IN,ACC,OUT>
    IN为输入流的元素类型，这里为Tuple<String,Integer> 表示一个用户和消费金额
    ACC为累加器类型（也就是计算时中间结果类型），这里要 保存用户名、累加消费金额和次数，所以是Tuple3<String,Double,Integer>
    OUT为返回结果类型，这里要将结果拼接成字符串，所以是String
*/
class MyAggregateFunction implements AggregateFunction<Tuple2<String,Integer>, Tuple3<String,Double,Integer>,String>{

    //创建初始累加器
    @Override
    public Tuple3<String, Double, Integer> createAccumulator() {
        return Tuple3.of("",0.0,0);
    }

    @Override
    //value是流上的元素，将value累加到accumulator中
    public Tuple3<String, Double, Integer> add(Tuple2<String, Integer> value, Tuple3<String, Double, Integer> accumulator) {
        accumulator.setFields(value.f0,accumulator.f1+value.f1,accumulator.f2+1);
        return accumulator;
    }

    @Override
    //从累加器中获取结果
    public String getResult(Tuple3<String, Double, Integer> accumulator) {
        return accumulator.f0+"消费了"+accumulator.f2+"次,平均消费"+(accumulator.f1/ accumulator.f2);
    }

    @Override
    //合并2个累加器，只有会话窗口才会触发
    public Tuple3<String, Double, Integer> merge(Tuple3<String, Double, Integer> a, Tuple3<String, Double, Integer> b) {
        return Tuple3.of(a.f0,a.f1+b.f1,a.f2+b.f2);
    }
}